package com.moli.iov.gateway.server;


import com.moli.iov.gateway.common.NamedThreadFactory;
import com.moli.iov.gateway.util.Partitioner.Partitioner;
import com.moli.iov.gateway.util.Partitioner.VinQueuePartitioner;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import org.slf4j.LoggerFactory;

/**
 * vin分区线程池
 *
 * @author lws
 * @version V1.0
 * @since JDK1.8 2017年10月19日
 */
public class VinThreadPoolExecutor {
    private static org.slf4j.Logger logger = LoggerFactory.getLogger(VinThreadPoolExecutor.class);
    /**
     * 线程数量
     */
    private int nThreads = Runtime.getRuntime().availableProcessors() * 2;
    /**
     * 线程池名称前缀
     */
    private String threadPoolPrefix = "";
    /**
     * 线程工厂
     */
    private ThreadFactory threadFactory;
    /**
     * 分区算法
     */
    private Partitioner partitioner;

    /**
     * 线程LIST
     */
    private List<Worker> workers = null;

    /**
     * 线程池队列大小
     */
    private Integer quenueSize = Integer.MAX_VALUE;

    /**
     * 初始化
     */
    private void init() {
        this.threadFactory = new NamedThreadFactory(threadPoolPrefix);
        this.partitioner = new VinQueuePartitioner();
        this.workers = new ArrayList<>(nThreads);
        for (int i = 0; i < this.nThreads; i++) {
            Worker worker = new Worker(new LinkedBlockingQueue<>(quenueSize));
            workers.add(worker);
            Thread thread = this.threadFactory.newThread(worker);
            thread.start();
        }
    }

    /**
     * 执行方法
     *
     * @param key
     * @param command
     */
    public void execute(Object key, Runnable command) {
        boolean offerResult = false;
        int partition = this.partitioner.partition(key, this.nThreads);
        //无阻塞的方式，防止阻塞主线程
        BlockingQueue<Runnable> workQueue = this.workers.get(partition).getWorkQueue();
        try {
            offerResult = workQueue.offer(command);
        } catch (Exception e) {
            logger.error(threadPoolPrefix + " Error", e);
        }
        if (!offerResult) {
            // TODO: 2019/5/17  打印丢弃的数据
            logger.error(threadPoolPrefix + "队列已满,size={}", workQueue.size());
        }
    }

    /**
     * 工作线程
     *
     * @author lws
     * @version V1.0
     * @since JDK1.8 2017年10月19日
     */
    class Worker implements Runnable {
        private BlockingQueue<Runnable> workQueue;

        public Worker(BlockingQueue<Runnable> workQueue) {
            this.workQueue = workQueue;
        }

        @Override
        public void run() {
            Runnable command = null;
            while (true) {
                try {
                    if ((command = workQueue.take()) != null) {
                        command.run();
                    }

                } catch (Exception e) {
                    logger.error("工作线程处理异常", e);
                }
            }

        }

        public BlockingQueue<Runnable> getWorkQueue() {
            return workQueue;
        }

    }


    public int getnThreads() {
        return nThreads;
    }

    public void setnThreads(int nThreads) {
        this.nThreads = nThreads;
    }

    public String getThreadPoolPrefix() {
        return threadPoolPrefix;
    }

    public void setThreadPoolPrefix(String threadPoolPrefix) {
        this.threadPoolPrefix = threadPoolPrefix;
    }

    public Integer getQuenueSize() {
        return quenueSize;
    }

    public void setQuenueSize(Integer quenueSize) {
        this.quenueSize = quenueSize;
    }

}
